[kafka] Cover Produce lifecycle and replication regressions - #4299

Queue a separate fetch completion for each successfully appended bucket so
an ordinary exception cannot prevent other buckets from being checked.
Keep each queue drain bounded and allow fatal errors to propagate.

Cover partial append failures, multiple buckets, queue execution behavior,
and RPC completion timing. Include the delayed fetch wakeup analysis.

Validate RPC and server unit and integration suites, including the 44
focused regression tests. The existing disk usage equality test passed
in isolation after a transient failure while reading live disk space.
Checkstyle, Spotless, license checks, and packaging also pass.

Co-Authored-By: Codex <noreply@openai.com>
AI-Model: gpt-6-astra
AI-Contributed/Feature: 911/983
AI-Contributed/UT: 0/339

Merge Apache main at 8a381bd and migrate the remaining fetch-result
constructor call introduced by this PR to FetchLogResultForBucket.records.
Use -1 for the unset filtered and retention offsets.

Validate the merged source with Java 11 regression tests and a clean
Java 8 compilation of the RPC/server modules and their test sources.

Co-Authored-By: Codex <noreply@openai.com>
AI-Model: gpt-6-astra
AI-Contributed/Feature: 0/9659
AI-Contributed/UT: 6/4948

Catch failures while mapping asynchronous handler results so the dispatcher
future completes exceptionally instead of remaining pending. Cover handler
futures that fail before and after dispatch registers its callback.

Validated with Java 11: Maven reactor build and 17 targeted Kafka tests,
including Checkstyle, Spotless, and license checks.

Co-Authored-By: Codex <noreply@openai.com>
AI-Model: gpt-6
AI-Contributed/Feature: 27/27
AI-Contributed/UT: 28/28

Route requests through the dispatcher and advertise only implemented APIs.
Return version-aware errors for unsupported APIs and invalid requests.

Validated with mvn -o -pl fluss-kafka verify (23 unit tests and 1 IT).

Co-Authored-By: Codex <noreply@openai.com>
AI-Model: gpt-6
AI-Contributed/Feature: 313/313
AI-Contributed/UT: 137/137

Extract topic identity and the raw/string table mapping contract before Metadata. Validate table kinds, field projections and metadata columns independently of request handling and record decoding.

Validation: Java 11, mvn -o -pl fluss-kafka clean verify (47 unit tests and 2 integration tests); Checkstyle, Spotless and RAT passed.

Co-Authored-By: Codex <noreply@openai.com>
AI-Model: gpt-6
AI-Contributed/Feature: 769/769
AI-Contributed/UT: 480/480

Resolve database.table topic names directly to Fluss table paths and return
qualified names for metadata. Validate name boundaries and cover same-named
tables in different databases through native metadata round trips.

Read projected field names directly from DataField to avoid rebuilding the
complete field-name list for every projected column.

Validated on Java 11: 67 Kafka unit tests and 2 integration tests passed,
along with Checkstyle, Spotless, RAT, and git diff --check.

Co-Authored-By: Codex <noreply@openai.com>
AI-Model: gpt-6
AI-Contributed/Feature: 52/52
AI-Contributed/UT: 167/167

Integrate the DDL mapping prerequisite and use its resolver for Metadata discovery. Omit unsupported tables from all-topic queries and return per-topic mapping errors for named queries. Preserve metadata for compatible tables in mixed requests.

Validation: Java 11, mvn -o -pl fluss-kafka clean verify (62 unit tests and 2 integration tests); Checkstyle, Spotless and RAT passed.

Co-Authored-By: Codex <noreply@openai.com>
AI-Model: gpt-6
AI-Contributed/Feature: 742/742
AI-Contributed/UT: 571/571

Route Metadata queries through the database.table mapping from PR03 and
enumerate compatible topics across databases. Preserve brokers and valid
topics when a requested database is absent or disappears during discovery;
propagate other gateway failures.

Cover qualified names, same-named tables across databases, synchronous and
asynchronous lookup failures, and Metadata v0-v11 over a real listener.

Validated on Java 11: 90 unit tests and 4 integration tests passed, along
with Checkstyle, Spotless, RAT, and git diff --check. Five additional
interleaved asynchronous review probes passed.

Co-Authored-By: Codex <noreply@openai.com>
AI-Model: gpt-6
AI-Contributed/Feature: 129/129
AI-Contributed/UT: 514/514

Validate non-idempotent Produce v3-v11 and isolate invalid partitions before invoking a protocol-independent backend. Preserve partition order, acknowledgements and owned record data. Leave production registration to the append integration PR.

Validation: Java 11, mvn -o -pl fluss-kafka spotless:apply clean verify; 69 unit tests and 2 integration tests passed, including Checkstyle, Spotless and RAT.

Co-Authored-By: Codex <noreply@openai.com>
AI-Model: gpt-6
AI-Contributed/Feature: 621/621
AI-Contributed/UT: 366/366

Read compressed records through a closeable streaming iterator to avoid
preallocating a record list from the declared batch count. Reject incoming
Kafka batches whose offset range disagrees with their record count before
copying records or invoking the backend. Keep Fluss-assigned offsets independent
of this input validation and preserve partition error isolation.

Cover all five codecs, owned record copies, malformed counts and offset ranges,
and backend-assigned offsets. Java 11 validation passed: 117 unit tests and
4 integration tests, plus Checkstyle, Spotless, RAT, and git diff --check.

Co-Authored-By: Codex <noreply@openai.com>
AI-Model: gpt-6
AI-Contributed/Feature: 42/42
AI-Contributed/UT: 168/168

Reuse the DDL contract to assemble key, value, timestamp and ordered headers into physical rows. Decode strings with strict UTF-8 validation and preserve nullable values. Return owned heap log bytes and recycle Arrow writers on success and failure.

Validation: Java 11, mvn -o -pl fluss-kafka spotless:apply clean verify; 75 unit tests and 2 integration tests passed, including Checkstyle, Spotless and RAT.

Co-Authored-By: Codex <noreply@openai.com>
AI-Model: gpt-6
AI-Contributed/Feature: 314/314
AI-Contributed/UT: 326/326

Decode and append one record at a time instead of retaining an entire row
list. Borrow command-owned key, value and header bytes during synchronous
conversion while preserving defensive-copy accessors.

Cover row and payload reuse, batches beyond initial Arrow capacity, and
resource cleanup after partial conversion failures.

Validation on Java 11: reactor clean verify passed with 130 unit tests and
4 integration tests, plus Checkstyle, Spotless, RAT and git diff --check.
An additional 25 boundary scenarios passed.

Co-Authored-By: Codex <noreply@openai.com>
AI-Model: gpt-6
AI-Contributed/Feature: 101/101
AI-Contributed/UT: 224/224

Base PR07 on PR06 04a9093, including the qualified topic contract,
Produce validation fixes, and streaming raw/string transcoding.
Retain independent core PR apache#3412 at 969f985 without changing its patch.

Co-Authored-By: Codex <noreply@openai.com>
AI-Model: gpt-6
AI-Contributed/Feature: 0/1046
AI-Contributed/UT: 0/309

Register Produce v3-v11 and adapt the local TabletServer gateway using
the shared qualified-topic DDL contract and raw/string transcoder.
Preserve the resolved TablePath across metadata lookup and conversion,
isolate table/partition failures, and map invalid topic names correctly.
Drain deferred actions immediately after the synchronous append invocation
so acks=all does not wait for its own delayed replica fetches.

Base on PR06 04a9093 and independent core PR apache#3412 at 969f985.

Validation: Java 11 reactor verify; 139 Kafka unit tests, 7 Kafka integration
tests, and 10 targeted RPC/server tests passed. Three-replica tests cover
raw/string, gzip, acks=0/1/all, offsets, timestamp/headers, native readback,
and independent writes to same-named tables in different databases.
Checkstyle, Spotless, RAT, and git diff --check passed.

Co-Authored-By: Codex <noreply@openai.com>
AI-Model: gpt-6
AI-Contributed/Feature: 319/319
AI-Contributed/UT: 606/606

Reuse parsed table metadata and immutable raw/string write plans while
refreshing native metadata and authorization for every topic write.
Offload metadata lookup, conversion and native submission to bounded FIFO
workers, preserving connection order without waiting for acknowledgements.
Close protocol-owned workers after transport and RPC processors stop.

Validation: Java 11 reactor verify passed 160 Kafka unit tests, 8 Kafka
integration tests and 5 NettyServer tests. Checkstyle, Spotless, RAT and
git diff --check passed. Tests cover cache invalidation, authorization
refresh, saturation, shutdown, pending acknowledgements and pipelined
three-replica append order. Source content is unchanged since validation.

ZooKeeper query reduction and byte-based admission remain follow-up work.

Co-Authored-By: Codex <noreply@openai.com>
AI-Model: gpt-6
AI-Contributed/Feature: 812/812
AI-Contributed/UT: 1047/1047

Co-Authored-By: Codex <noreply@openai.com>
AI-Model: gpt-6
AI-Contributed/Feature: 0/0
AI-Contributed/UT: 364/364

Adapt Produce regressions to database-qualified topic names from PR07.
Exercise a real leader handoff while the old node stays online, verify
NOT_LEADER_OR_FOLLOWER on the existing connection, and confirm the same
producer refreshes its route without disconnecting. Check acks=all offsets
and native readback across the handoff.

Co-Authored-By: Codex <noreply@openai.com>
AI-Model: gpt-6
AI-Contributed/Feature: 0/0
AI-Contributed/UT: 285/285